package com.shujia.base.javamr;

import java.io.*;
import java.util.HashMap;

public class Map {
    public static void main(String[] args) throws Exception{
        File file = new File("hadoop/data/blocks");

        //获取当前目录下的文件
        String[] blocks = file.list();

        if(blocks!=null){
            for (String block : blocks) {
                //将每一个文件封装成一个线程，计算当前block块中的文件数量
                new MapTask(block).start();
            }
        }
    }
}

class MapTask extends Thread{
    private String blockName;
    final BufferedReader br;
    final BufferedWriter bw;

    public MapTask() throws Exception{
        br = null;
        bw = null;
    }

    public MapTask(String blockName) throws Exception{
        br = new BufferedReader(new FileReader("hadoop/data/blocks/" + blockName));
        bw = new BufferedWriter(new FileWriter("hadoop/data/maps/part-m-" + blockName.split("---")[1]));
        this.blockName = blockName;
    }

    @Override
    public void run() {
        try {
            HashMap<String, Long> map = new HashMap<>();
            String line = null;
            while ((line=br.readLine())!=null){
                String clazz = line.split(",")[4];
                if(map.containsKey(clazz)){
                    map.put(clazz,map.get(clazz)+1);
                }else {
                    map.put(clazz,1L);
                }
            }

            //将当前block块计算的结果写入到一个文件
            map.forEach((clazz,number)->{
                try {
                    bw.write(clazz+"\t"+number);
                    bw.newLine();
                    bw.flush();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try {
                bw.close();
                br.close();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}
